﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Threading.Actors.Runtime;
using System.Fabric;
using System.Globalization;
using System.Threading.Tasks;
using System.Threading.Actors.Languages;
using System.Threading.Actors.Utilities;
using System.Collections.Generic;
using System.Collections.Concurrent;

namespace System.Threading.Actors
{
    public class ReplicatingActorServiceBase : StatefulServiceReplica, IActor
    {
        private Uri listenUri;
        private ActorRawTcpListener listener;

        protected ReplicatingActorState stateProvider;

        protected ActorServiceHelper m_defaultHelper;
        // Make this a ConcurrentDictionary because: (1) it may be updated by many threads
        // concurrently, and (2) it may be accessed/traversed while it is being updated.
        protected ConcurrentDictionary<string, ActorServiceHelper> m_keySpecificHelpers;

        // Mirrors WSAEADDRINUSE in WinError.h
        private const int AddressInUse = 10048;
		
        private Action<IActorState> m_methodInitializer;

        public ReplicatingActorServiceBase(Action<IActorState> derivedClassInitializer)
        {
            m_methodInitializer = (astate) => {
                astate.Set("Ping", (Func<IActorState,object[],object>)delegate(IActorState state,object[] args) 
                {
                    return state.GetId();
                });
                if (derivedClassInitializer != null) derivedClassInitializer(astate);
            };

            m_keySpecificHelpers = new ConcurrentDictionary<string, ActorServiceHelper>();
        }

#region StatefulServiceReplica methods

        protected override ReplicatorSettings CreateReplicatorSettings()
        {
            // Provide replicator settings
            var endpoint = this.InitializationParameters.CodePackageActivationContext.GetEndpoints()["ActorServiceEndPoint_Replicator"];
            return new ReplicatorSettings
            {
                ReplicatorAddress = string.Format(CultureInfo.InvariantCulture, "{0}:{1}", NetworkConfiguration.GetIPAddressOrHostName(true, NetworkConfiguration.PreferIPv6), endpoint.Port),
                BatchAcknowledgementInterval = TimeSpan.FromMilliseconds(0),   // Eliminate forced delay between replication start and end
                // RetryInterval = TimeSpan.FromMilliseconds(500) -- Doesn't seem to help the 10-sec service startup problem
                
            };
        }

        // Grab an ActorServiceHelper given a partition key.
        internal ActorServiceHelper GetServiceHelper(string partitionName)
        {
            if (String.IsNullOrEmpty(partitionName))
            {
                // No partition key means that we use the default service helper.
                if (m_defaultHelper == null)
                {
                    bool created = false;
                    lock (m_keySpecificHelpers)
                    {
                        if (m_defaultHelper == null)
                        {
                            var isolatedActorState = stateProvider.GetIsolatedState(partitionName); // escapes the lock by doing await-related stuff on another Task
                            m_defaultHelper = new ActorServiceHelper(isolatedActorState, this.InitializationParameters.ServiceName.ToString(), m_methodInitializer);
                            created = true;
                        }
                    }
                    if (created)
                        ActorETWEventSource.Log.ActorCreated(stateProvider.GetId(), null);
                }
                return m_defaultHelper;
            }
            else
            {
                // The presence of a partition key means that we need to use the service helper that
                // corresponds to that partition key.
                ActorServiceHelper result;
                if(!m_keySpecificHelpers.TryGetValue(partitionName, out result))
                {
                    // I use double-check locking even though m_keySpecificHelpers is a ConcurrentDictionary,
                    // because I don't trust ConcurrentDictionary.GetOrAdd to do the right thing
                    // without explicit locking around it.
                    bool created = false;
                    lock (m_keySpecificHelpers)
                    {
                        if (!m_keySpecificHelpers.TryGetValue(partitionName, out result))
                        {
                            var isolatedActorState = stateProvider.GetIsolatedState(partitionName); // escapes the lock by doing await-related stuff on another Task
                            result = new ActorServiceHelper(isolatedActorState, this.InitializationParameters.ServiceName.ToString(), m_methodInitializer);
                            m_keySpecificHelpers[partitionName] = result;
                            created = true;
                        }
                    }
                    if (created)
                        ActorETWEventSource.Log.ActorCreated(stateProvider.GetId(), partitionName);
                }
                return result;
            }

        }

        protected override void OnOpen(IStatefulServicePartition partition)
        {
            // reading the endpoint from what the baseclass provided
            var endpoints = this.InitializationParameters.CodePackageActivationContext.GetEndpoints();
            var ep = endpoints["ActorServiceEndPoint"];

            // figure out the listen URI
            // note - using the partitionId and replicaId helps ensure uniqueness if multiple partitions are placed on the same node
            this.listenUri = new UriBuilder(Uri.UriSchemeNetTcp, NetworkConfiguration.GetIPAddressOrHostName(false, NetworkConfiguration.PreferIPv6), ep.Port, this.InitializationParameters.PartitionId.ToString() + "_" + this.InitializationParameters.ReplicaId.ToString(CultureInfo.InvariantCulture)).Uri;

            // setup the listener
            this.Logger.Write(Guid.Empty, "OnOpen: Setting up listener at {0} for partition {1}", this.listenUri, partition.PartitionInfo.Id);
        }

        protected override string GetEndpoint()
        {
            return this.listenUri.ToString();
        }

        protected override void StartListening()
        {
            this.Logger.Write(Guid.Empty, "Starting listener");
            
            // Register ourself with the global listener for this port
            try
            {
                this.listener = ActorRawTcpListener.GetListener(listenUri);
                listener.RegisterService(listenUri, this);
            }
            catch (Exception e)
            {
                Console.WriteLine("StartListening({0}) threw an exception: {1}", listenUri, e);
                throw;
            }

            // We need to do the OnPromotionToPrimary operations (which involve "resurrecting"
            // proxies and subscriptions) separately from the main StartListening logic.  Otherwise,
            // two actors that are subscribed to each other and simultaneously fail over could 
            // end up deadlocked, as neither of their StartListening methods (which establish a listener)
            // could successfully complete.
            Task.Factory.StartNew(spo =>
                {
                    var sp = spo as ReplicatingActorState;
                    lock (sp) sp.OnPromotionToPrimary(); // Needs to be under lock
                }, stateProvider)
            .ContinueWith(antecedent =>
                {
                    // Let's observe any exceptions thrown from OnPromotionToPrimary
                    Console.WriteLine("OnPromotionToPrimary failed: {0}", antecedent.Exception);
                    this.Logger.Write(Guid.Empty, "OnPromotionToPrimary failed: {0}", antecedent.Exception);
                }, TaskContinuationOptions.OnlyOnFaulted);
        }

        protected override void StopListening()
        {
            this.Logger.Write(Guid.Empty, "Stopping listener");
            this.listener.UnregisterService(this.listenUri);
        }

        protected override void OnClose()
        {
            m_defaultHelper = null;
            m_keySpecificHelpers = null;
            if (stateProvider != null) stateProvider.Dispose();
            if(listener != null) listener.UnregisterService(this.listenUri);
            this.listener = null;
        }

        protected override void OnAbort()
        {
            OnClose(); // Same logic
        }

        protected override StateProviderBase CreateStateProvider(IStatefulServicePartition partition)
        {
            var actorName = this.InitializationParameters.ServiceName.ToString();
            // The ReplicatingActorState constructor may encounter a number of await points, and this method
            // is called under lock.  So let's call the ReplicatingActorState constructor on a separate
            // thread in order to escape the lock.
            this.stateProvider = Task.Factory.StartNew(() =>
                {
                    return new ReplicatingActorState(this.Logger, actorName);
                }, TaskCreationOptions.PreferFairness).Result;

            return this.stateProvider;
        }

#endregion StatefulServiceReplica methods

        #region IActor implementation

        // Support methods from non-.NET languages
        public Task<string> AddMethod(int clientSequenceNumber, string partitionName, int language, string methodName, string methodImpl)
        {
            return GetServiceHelper(partitionName).AddMethod(clientSequenceNumber, (SupportedLanguages)language, methodName, methodImpl);
        }

        public Task<string> AddAssembly(int clientSequenceNumber, string partitionName, string assemblyName, byte[] assemblyBytes)
        {
            return GetServiceHelper(partitionName).AddAssembly(clientSequenceNumber, assemblyName, assemblyBytes);
        }

        public Task<string> CallMethod(string clientId, int clientSequenceNumber, string partitionName, string methodName, string[] parameters)
        {
            return GetServiceHelper(partitionName).CallMethod(clientId, clientSequenceNumber, methodName, parameters);
        }

        public string Subscribe(int clientSequenceNumber, string partitionName, string eventType, IPublicationCallbackContract context)
        {
            return GetServiceHelper(partitionName).Subscribe(clientSequenceNumber, eventType, context);
        }

        public string Unsubscribe(int clientSequenceNumber, string partitionName, string eventType, IPublicationCallbackContract context)
        {
            return GetServiceHelper(partitionName).Unsubscribe(clientSequenceNumber, eventType, context);
        }

        public string UnsubscribeAll(int clientSequenceNumber, string partitionName, IPublicationCallbackContract context)
        {
            return GetServiceHelper(partitionName).UnsubscribeAll(clientSequenceNumber, context);
        }

        public string Ping(int clientSequenceNumber, string partitionName)
        {
            var myAddress = this.InitializationParameters.ServiceName.ToString();
            return ActorServiceResult.FromResult(clientSequenceNumber, myAddress).ToJson();
        }

        public string Configure(int clientSequenceNumber, string partitionName, string configurationString)
        {
            return GetServiceHelper(partitionName).Configure(clientSequenceNumber, configurationString);
        }

        public string DeleteSelf(int clientSequenceNumber, string partitionName)
        {
            String ret = GetServiceHelper(partitionName).DeleteSelf(clientSequenceNumber);
            if (partitionName == null)
            {
                m_defaultHelper = null;
            }
            else
            {
                lock (m_keySpecificHelpers)
                {
                    ActorServiceHelper serviceHelper = null;
                    if (!m_keySpecificHelpers.TryRemove(partitionName, out serviceHelper))
                    {
                        // We lost a race to remove this.  Let's assume this is a safe race to lose.
                    }
                }
            }
            return ret;
        }

        #endregion
    }

}
